package com.hliushi.utils

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.sql.{ForeachWriter, Row}

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/7/2 15:09
 */
abstract class BasicJdbcSink(sql: String) extends ForeachWriter[Row] {

  var conn: Connection = _
  var ps: PreparedStatement = _

  // 开启连接
  override def open(partitionId: Long, version: Long): Boolean = {
    conn = DriverManager.getConnection("jdbc:mysql://cvm-node01:3306/bigdata?characterEncoding=UTF-8&useSSL=false",
      "root", "123456")
    true
  }

  // 处理数据--将数据存入到MySQL
  override def process(row: Row): Unit = {
    realProcess(sql, row)
  }

  def realProcess(sql: String, row: Row): Unit

  // 关闭连接
  override def close(errorOrNull: Throwable): Unit = {
    if (ps != null) {
      ps.close()
    }
    if (conn != null) {
      conn.close()
    }
  }


}